package com.gitee.wsl.jvm.okio

import android.annotation.SuppressLint
import android.os.Bundle
import android.os.Handler
import android.os.Message
import android.util.Log
import com.gitee.wsl.jvm.okio.OkioSocket.OnSendCallBack
import okio.BufferedSink
import okio.BufferedSource
import okio.buffer
import okio.sink
import okio.source
import java.io.Closeable
import java.io.IOException
import java.net.InetSocketAddress
import java.net.Socket
import java.net.SocketException
import java.net.SocketTimeoutException
import java.nio.charset.Charset
import java.text.MessageFormat
import java.util.Arrays
import java.util.Deque
import java.util.LinkedList
import java.util.concurrent.TimeUnit


/**
 * Created by dab on 2017/9/8 0008 17:05
 */
class OkioSocket @SuppressLint("HandlerLeak") constructor() : Closeable {
    private var mSink: BufferedSink? = null
    private var mSource: BufferedSource? = null
    private var readThread: Thread? = null
    private var writeThread: Thread? = null
    private val msgQueue //消息队列
            : TaskQueue<Request> = TaskQueue()
    private var mOnConnectCallBack: OnConnectCallBack? = null
    private var mOnMessageChange: OnMessageCome? = null
    private var mOnEncryption: OnEncryption? = null
    private var mOnDecode: OnDecode? = null
    private var canConnected = true //是否允许连接
    private var isConnect = false //是否连接成功
    private var isConnecting = false //连接中
    private var isFirstConnected = true //第一次连接
    private var reConnectedCount = 0 //当前重连次数
    private var mHandler: Handler
    private var isClose = false // 是否关闭

    /**
     * 设置是否可以连接(马上会尝试连接)
     *
     * @param canConnected
     */
    fun setCanConnected(canConnected: Boolean): OkioSocket {
        this.canConnected = canConnected
        reConnected()
        return this
    }

    /**
     * 设置解密
     *
     * @param onDecode
     */
    fun setOnDecode(onDecode: OnDecode?): OkioSocket {
        mOnDecode = onDecode
        return this
    }

    /**
     * 设置加密
     *
     * @param onEncryption
     */
    fun setOnEncryption(onEncryption: OnEncryption?): OkioSocket {
        mOnEncryption = onEncryption
        return this
    }

    fun setCharset( charset: Charset): OkioSocket {
        Configs.charset = charset
        return this
    }

    /**
     * 设置连接的监听
     *
     * @param onConnectCallBack
     * @return
     */
    fun setOnConnectCallBack(onConnectCallBack: OnConnectCallBack?): OkioSocket {
        mOnConnectCallBack = onConnectCallBack
        return this
    }

    /**
     * 设置消息的监听
     *
     * @param onMessageChange
     * @return
     */
    fun setOnMessageChange(onMessageChange: OnMessageCome?): OkioSocket {
        mOnMessageChange = onMessageChange
        return this
    }

    /**
     * 设置重新连接的间隔(ms) 最少1000,否则为默认4000
     *
     * @param reConnectedTime
     */
    fun setReConnectedTime(reConnectedTime: Long): OkioSocket {
        if (reConnectedTime >= 1000) {
            Configs.reConnectedTime = reConnectedTime
        }
        return this
    }

    /**
     * 设置连接Socket超时的时间(ms),最少1000,否则为默认8000
     *
     * @param connectTimeout
     */
    fun setConnectTimeout(connectTimeout: Int): OkioSocket {
        if (connectTimeout >= 1000) {
            Configs.connectTimeout = connectTimeout
        }
        return this
    }

    /**
     * 设置读的超时(读超时以后,暂时会被当成断开连接处理,以后需要了再修改)
     *
     * @param readTimeout
     */
    private fun setReadTimeout(readTimeout: Long): OkioSocket {
        if (readTimeout >= 1000) {
            Configs.readTimeout = readTimeout
        }
        return this
    }

    /**
     * 设置写的超时(写超时以后,会放弃这个任务,并且会回调OnSendCallBack,如果发送消息时设置了的话)
     *
     * @param writeTimeout
     */
    fun setWriteTimeout(writeTimeout: Long): OkioSocket {
        if (writeTimeout >= 1000) {
            Configs.writeTimeout = writeTimeout
        }
        return this
    }

    init {
        mHandler = object : Handler() {
            override fun handleMessage(msg: Message) {
                super.handleMessage(msg)
                when (msg.what) {
                    Configs.MessageWhat.CONNECTED -> if (mOnConnectCallBack != null) {
                        val isConnected: Boolean = msg.getData().getBoolean("isConnected", false)
                        val err = msg.getData().getSerializable("err") as IOException
                        mOnConnectCallBack!!.onConnect(isConnected, isFirstConnected, err)
                    }

                    Configs.MessageWhat.RE_CONNECTED -> reConnected()
                    Configs.MessageWhat.DATA -> {
                        var data: ByteArray? = msg.getData().getByteArray("data")
                        if (data == null) {
                            Configs.e("data ==nul")
                            return
                        }
                        if (mOnDecode != null) {
                            data = mOnDecode!!.onDecode(data)
                        }
                        val message = String(data, Configs.charset)
                        Configs.e("<<<<<<<<$message")
                        if (mOnMessageChange != null) {
                            mOnMessageChange!!.onMessageCome(message)
                        }
                    }
                }
            }
        }
    }

    /**
     * 链接,如果已经在连接成功了或者正在连接中,就不再会连接接了
     *
     * @param host
     * @param port
     */
    fun connect( host: String, port: Int) {
        if (!isConnect && !isConnecting) {
            isClose = false
            connectSocket(host, port)
        }
    }

    private fun connectSocket(host: String, port: Int) {
        Configs.HOST = host
        Configs.PORT = port
        synchronized(OkioSocket::class.java) {
            if (isConnecting) return
            isConnecting = true
            //这里应该可以用线程池来减少多次重连的创建县线程
            readThread = Thread(connectService())
            readThread!!.name = "readThread:" + readThread!!.id
            readThread!!.start()
        }
    }

    /**
     * 重新链接
     */
    fun reConnected() {
        Configs.e(
            "尝试连接" +
                    "\n连接断开了:" + !isConnect +
                    "\n没有正在重连:" + !isConnecting +
                    "\n没有关闭:" + !isClose +
                    "\n允许重连:" + canConnected
        )
        if (isConnect) return
        if (isConnecting) return
        if (isClose) return
        if (!canConnected) return
        Configs.HOST?.let { connectSocket(it, Configs.PORT) }
    }

    /**
     * 发送一个消息(添加到队列),不关心发送成功或者失败
     *
     * @param sendMsg
     */
    fun send(sendMsg: String) {
        send(sendMsg, null)
    }

    /**
     * 发送一个消息(添加到队列,并且监听发送成功或者失败)
     *
     * @param sendMsg
     */
    fun send( sendMsg: String, onSendCallBack: OnSendCallBack?): Request? {
        if (isClose) {
            onSendCallBack?.onSendCallBack(RuntimeException("已经关闭或者未链接成功"))
            return null
        }
        val request = Request(sendMsg, onSendCallBack)
        msgQueue.addTask(request)
        if (writeThread == null) {
            writeThread = Thread(sendService())
            writeThread!!.name = "writeThread:" + writeThread!!.id
            writeThread!!.start()
        }
        return request
    }

    /**
     * 移除请求
     *
     * @param requests
     */
    fun remove(vararg requests: Request) {
        if (requests == null) return
        for (request: Request in requests) {
            request.clear()
            msgQueue.remove(request)
        }
    }

    /**
     * 移除请求
     *
     * @param requests
     */
    fun remove(requests: List<Request>?) {
        if (requests == null) return
        for (request: Request in requests) {
            request.clear()
            msgQueue.remove(request)
        }
    }

    @Throws(IOException::class)
    override fun close() {
        isClose = true
        isConnect = false
        msgQueue.clear()
        setCanConnected(false)
        if (mHandler != null) {
            mHandler.removeCallbacksAndMessages(null)
        }
        if (mSink != null) {
            mSink!!.close()
        }
        if (mSource != null) {
            mSource!!.close()
        }
    }

    /**
     * 链接和读消息
     */
    private inner class connectService() : Runnable {
        override fun run() {
            try {
                Configs.e("正在连接:" + Thread.currentThread().name)
                val socket = Socket()
                socket.connect(
                    InetSocketAddress(Configs.HOST, Configs.PORT),
                    Configs.connectTimeout
                )
                mSink = socket.sink().buffer()
                mSource = socket.source().buffer()
                //                initSink(mSink);
//                initSource(mSource);
                sendMessage(Configs.MessageWhat.CONNECTED) { bundle: Bundle ->
                    bundle.putBoolean("isConnected", socket.isConnected())
                    bundle
                }
                isConnecting = false
                isFirstConnected = false
                reConnectedCount = 0
                isConnect = true
                receiveMsg()
            } catch (e: IOException) {
                e.printStackTrace()
                isConnecting = false
                isConnect = false
                sendMessage(Configs.MessageWhat.CONNECTED, OnSetData { bundle: Bundle ->
                    bundle.putSerializable("err", e)
                    bundle
                })
                sendReConnectedMessage()
                reConnectedCount++
                Configs.e("当前重连次数$reConnectedCount")
            }
        }
    }

    /**
     * 初始化Source
     *
     * @param source
     */
    private fun initSource(source: BufferedSource?) {
        if (source == null) return
        if (Configs.readTimeout < 1000) return
        val timeout = source.timeout()
        timeout.deadline(Configs.readTimeout, TimeUnit.MILLISECONDS)
    }

    /**
     * 初始化Sink
     *
     * @param sink
     */
    private fun initSink(sink: BufferedSink?) {
        if (sink == null) return
        if (Configs.writeTimeout < 1000) return
        val timeout = sink.timeout()
        timeout.deadline(Configs.writeTimeout, TimeUnit.MILLISECONDS)
    }

    /**
     * 读消息
     */
    private fun receiveMsg() {
        Configs.e("链接成功:" + Thread.currentThread().name)
        //        send("dasdasdsadada");
        try {
            val timeout = mSource!!.timeout()
            while (!isClose) {
                timeout.clearDeadline()
                val length = mSource!!.readInt()
                Log.e(TAG, "receiveMsg555555555: $length")
                timeout.deadline(Configs.readTimeout, TimeUnit.MILLISECONDS)
                val bytes = mSource!!.readByteArray(length.toLong())
                Log.e(TAG, "receiveMsg: " + Arrays.toString(bytes))
                sendMessage(Configs.MessageWhat.DATA, OnSetData { bundle: Bundle ->
                    bundle.putByteArray("data", bytes)
                    bundle
                })
            }
        } catch (e: IOException) {
            e.printStackTrace()
            sendMessage(Configs.MessageWhat.CONNECTED, OnSetData { bundle: Bundle ->
                bundle.putSerializable("err", e)
                bundle
            })
            isConnect = false
            if (e is SocketException || e is SocketTimeoutException) {
                sendReConnectedMessage()
            }
        }
    }

    /**
     * 发送消息
     */
    private inner class sendService() : Runnable {
        override fun run() {
            while (!isClose && mSink != null) {
                val request = msgQueue.popTask()
                if (request == null) {
                    Configs.e(">>>>>>>>>> requestSucceed is null")
                    continue
                }
                val onSendCallBack= request.onSendCallBack
                try {
                    val msg: String = request.msg
                    var bytes = msg.toByteArray()
                    if (mOnEncryption != null) {
                        bytes = mOnEncryption!!.onEncryption(bytes)
                    }
                    mSink!!.writeInt(bytes.size)
                    mSink!!.write(bytes)
                    mSink!!.flush()
                    Configs.e("ThreadName:" + Thread.currentThread().name + ">>>>>>>>>>" + msg)
                    onSendCallBack?.onSendCallBack(null)
                } catch (e: IOException) {
                    e.printStackTrace()
                    onSendCallBack?.onSendCallBack(e)
                    sendMessage(Configs.MessageWhat.CONNECTED, OnSetData { bundle: Bundle ->
                        bundle.putSerializable("err", e)
                        bundle
                    })
                    if (e is SocketException) {
                        isConnect = false
                        sendReConnectedMessage()
                    }
                }
            }
        }
    }

    /**
     * 发送一个消息
     *
     * @param what
     * @return
     */
    private fun sendMessage(what: Int,  onSetData: OnSetData) {
        val message: Message = mHandler.obtainMessage()
        message.what = what
        val bundle = Bundle()
        message.setData(onSetData.setData(bundle))
        mHandler.sendMessage(message)
    }
    ////////////////////////////////////////////////////////////////////////////////////////////////////
    //定义接口
    /**
     * 发送消息
     */
    interface OnSendCallBack {
        fun onSendCallBack(e: Exception?)
    }

    /**
     * 发送一个重连的消息
     */
    private fun sendReConnectedMessage() {
        val msg: Message = mHandler.obtainMessage()
        msg.what = Configs.MessageWhat.RE_CONNECTED
        mHandler.sendMessageDelayed(msg, Configs.reConnectedTime)
    }

    /**
     * 发送一个连接的消息
     */
    private fun sendConnectedMessage() {
        val msg: Message = mHandler.obtainMessage()
        msg.what = Configs.MessageWhat.CONNECTED
        mHandler.sendMessageDelayed(msg, Configs.reConnectedTime)
    }

    fun interface OnSetData {
        /**
         * 内部通讯
         *
         * @param bundle
         * @return
         */
        fun setData(bundle: Bundle): Bundle?
    }

    interface OnConnectCallBack {
        /**
         * @param succeed          是否连接成功
         * @param isFirstConnected 是否是第一次连接,false表示重新连接的
         * @param e
         */
        fun onConnect(succeed: Boolean, isFirstConnected: Boolean, e: Exception?)
    }

    interface OnEncryption {
        fun onEncryption(original: ByteArray?): ByteArray
    }

    interface OnDecode {
        fun onDecode(original: ByteArray?): ByteArray
    }

    interface OnMessageCome {
        fun onMessageCome(msg: String?)
    }

    companion object {
        private val TAG = "OkioSocket"
    }
}

/**
 * Created by dab on 2017/9/12 0012 14:10
 */
object Configs {
    var deBug = true
    var myTag = "****"
    var reConnectedTime: Long = 4000 //重连间隔
    var connectTimeout = 8000 //连接超时时间
    var readTimeout = Long.MAX_VALUE //读超时时间
    var writeTimeout: Long = 10000 //写超时时间
    var charset = Charset.forName("UTF-8")
    var HOST: String? = null
    var PORT = 0
    fun e(message: String?) {
        if (!deBug) {
            return
        }
        val stackTrace = Throwable().stackTrace
        val className = stackTrace[1].fileName
        val methodName = stackTrace[1].methodName
        val lineNumber = stackTrace[1].lineNumber
        Log.e(
            className,
            MessageFormat.format("({0}:{1}){2}{3}", className, lineNumber, myTag, message)
        )
    }

    object MessageWhat {
        /**
         * 是否连接
         */
        const val CONNECTED = 0x01

        /**
         * 重新连接
         */
        const val RE_CONNECTED = 0x02

        /**
         * 数据
         */
        const val DATA = 0x03
    }
}


/**
 * Created by dab on 2017/9/11 0011 15:39
 */
class TaskQueue<T> internal constructor() {
    private val deque: Deque<T>

    init {
        deque = LinkedList()
    }

    /**
     * 添加一个任务
     *
     * @param task
     * @return
     */
    fun addTask( task: T): TaskQueue<*> {
        synchronized(deque) {
            deque.push(task)
            //deque.notifyAll()
        }
        return this
    }

    /**
     * 移除任务
     *
     * @param task
     * @return
     */
    fun remove(task: T): Boolean {
        synchronized(deque) { return deque.remove(task) }
    }

    /**
     * 取出第一个并移除
     * 如果没有,就阻塞线程
     *
     * @return
     */
    fun popTask(): T? {
        synchronized(deque) {
            while (deque.isEmpty()) {
                try {
                    //deque.wait()
                } catch (e: InterruptedException) {
                    e.printStackTrace()
                }
            }
            return deque.pollFirst()
        }
    }

    /**
     * 取出第一个并移除,
     * 如果没有,就阻塞,如果ms毫秒内仍然没有,就返回null
     *
     * @param ms
     * @return
     */
    fun popTask(ms: Long): T? {
        synchronized(deque) {
            if (deque.isEmpty()) {
                try {
                    //deque.wait(ms)
                } catch (e: InterruptedException) {
                    e.printStackTrace()
                }
            }
            return if (deque.isEmpty()) {
                null //time out
            } else deque.pollFirst()
        }
    }

    fun clear() {
        synchronized(deque) { deque.clear() }
    }
}

/**
 * Created by dab on 2017/9/27 0027 09:45
 */
class Request(var msg: String, var onSendCallBack: OnSendCallBack?) {

    /**
     * 清除掉回调,防止内存泄露
     */
    fun clear() {
        onSendCallBack = null
    }
}

